package com.hhf.rrd.usage.source;

import com.hhf.rrd.model.Student;
import com.hhf.rrd.utils.MysqlUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Arrays;
import java.util.Properties;

/**
 * 数据源
 *      开启socket端口：nc -l 9999
 *      查看端口占用: lsof -i :9999
 *
 * @author huanghaifeng15
 * @date 2022/2/10 07:03
 **/
public class SourceApp {
    public static void main(String[] args) throws Exception {
        // 获取上下文
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 获取socket源
        DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 9999);

        // 读取某一 文件流
        DataStreamSource<String> readTextFileStream = env.readTextFile("/data/wc/wc.data");

        // 固定集合的Stream
        DataStreamSource<String> collectionStream = env.fromCollection(Arrays.asList("a", "b"));

        // 自定义数据源
        DataStreamSource<Long> sourceFunStream = env.addSource(new SourceFunction<Long>() {

            private long count = 0L;
            private volatile boolean isRunning = true;

            @Override
            public void run(SourceContext<Long> ctx) throws Exception {
                int maxNum = 1000;
                while (isRunning && count < maxNum) {
                    ctx.collect(count);
                    count++;
                }
            }

            @Override
            public void cancel() {
                isRunning = false;
            }
        });

        DataStreamSource<Student> richSourceFunStream = env.addSource(new RichSourceFunction<Student>() {

            private Connection connection;
            private PreparedStatement statement;

            @Override
            public void open(Configuration parameters) throws Exception {
                connection = MysqlUtils.getConnection();
                assert connection != null;
                statement = connection.prepareStatement("select * from student");
            }


            @Override
            public void run(SourceContext<Student> ctx) throws Exception {
                ResultSet resultSet = statement.executeQuery();
                while (resultSet.next()) {
                    Student student = new Student();
                    String name = resultSet.getString(0);
                    student.setName(name);
                    int age = resultSet.getInt(1);
                    student.setAge(age);
                    int sex = resultSet.getInt(2);
                    student.setSex(sex);

                    ctx.collect(student);
                }
            }

            @Override
            public void cancel() {

            }

            @Override
            public void close() throws Exception {
                MysqlUtils.close(connection, statement);
            }
        });

        DataStreamSource<Student> parallelSourceFunStream = env.addSource(new ParallelSourceFunction<Student>() {
            @Override
            public void run(SourceContext<Student> ctx) throws Exception {

            }

            @Override
            public void cancel() {

            }
        });

        env.addSource(new RichParallelSourceFunction<Student>() {
            @Override
            public void open(Configuration parameters) throws Exception {

            }

            @Override
            public void run(SourceContext<Student> ctx) throws Exception {

            }

            @Override
            public void cancel() {

            }

            @Override
            public void close() throws Exception {

            }
        });

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092,localhost:9093");
        properties.setProperty("group.id", "test");
        DataStreamSource<String> kafkaSourceStream = env.addSource(new FlinkKafkaConsumer<>("test_topic", new SimpleStringSchema(), properties));

        // 执行任务
        env.execute("sourceApp");
    }
}
